package kevin.hbase.service;

import cc.chinagps.gateway.buff.GBossDataBuff;
import kevin.hbase.bean.HbaseGps;
import kevin.hbase.common.Order;
import kevin.hbase.constant.Constant;
import kevin.hbase.util.DateTimeUtil;
import kevin.hbase.util.HBaseKeyUtil;
import kevin.hbase.util.HexUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;

@Service
@Slf4j
public class HbaseService {
    private static final byte[] family = "gps".getBytes();
    private static final byte[] qualifier_base = "baseinfo".getBytes();
    private static final byte[] qualifier_content = "content".getBytes();
    private static final byte[] qualifier_callLetter = "callletter".getBytes();

    @Autowired
    private Connection hbaseConnection;

    public HbaseGps getLastGps(String deviceId, Long startTime, Long endTime) {
        try {
            if (startTime == null) {
                startTime = DateTimeUtil.getTimestampOfDateTime(DateTimeUtil.parseStringToDateTime("2019-01-01 00:00:00", Constant.DATETIMEFORMAT));
            }

            if (endTime == null) {
                endTime = new Date().getTime() + 10 * 60 * 1000;
            }
            ResultScanner rsscan = getScanner(deviceId, startTime, endTime, Order.DESC);
            Iterator<Result> iter = rsscan.iterator();
            if (iter.hasNext()) {
                Result rs = iter.next();
                byte[] baseData = rs.getValue(family, qualifier_base);
                byte[] content = rs.getValue(family, qualifier_content);
                GBossDataBuff.GpsBaseInfo baseinfo = GBossDataBuff.GpsBaseInfo.parseFrom(baseData);
                HbaseGps gpsRecord = HbaseGps.valueOf(baseinfo);
                gpsRecord.setCallLetter(deviceId);
                gpsRecord.setSource(HexUtil.byteToHexStr(content));
                return gpsRecord;
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        return null;
    }

    public HbaseGps getLastGps(String deviceId) {
        return getLastGps(deviceId, null, null);
    }

    public List<HbaseGps> getHistoryGps(String callLetter, String startTime, String endTime) throws Exception {
        List<HbaseGps> list = new ArrayList<>();
        long start = DateTimeUtil.getTimestampOfDateTime(DateTimeUtil.parseStringToDateTime(startTime, Constant.DATETIMEFORMAT));
        long end = DateTimeUtil.getTimestampOfDateTime(DateTimeUtil.parseStringToDateTime(endTime, Constant.DATETIMEFORMAT));
        ResultScanner rsscan = getScanner(callLetter, start, end, Order.ASC);
        Iterator<Result> iter = rsscan.iterator();
        while (iter.hasNext()) {
            Result rs = iter.next();
            byte[] baseData = rs.getValue(family, qualifier_base);
            GBossDataBuff.GpsBaseInfo baseinfo = GBossDataBuff.GpsBaseInfo.parseFrom(baseData);

            HbaseGps gps = HbaseGps.valueOf(baseinfo);
            gps.setCallLetter(callLetter);
            list.add(gps);
        }
        return list;
    }

    public void delGpsByTime(String deviceId, String startTime, String endTime) throws Exception {
        long start = DateTimeUtil.getTimestampOfDateTime(DateTimeUtil.parseStringToDateTime(startTime, Constant.DATETIMEFORMAT));
        long end = DateTimeUtil.getTimestampOfDateTime(DateTimeUtil.parseStringToDateTime(endTime, Constant.DATETIMEFORMAT));
        byte[] startKey = HBaseKeyUtil.getKey(deviceId, start);
        byte[] endKey = HBaseKeyUtil.getKey(deviceId, end);
        Table table = hbaseConnection.getTable(TableName.valueOf("t_history"));
        Scan scan = new Scan(startKey, endKey);
        ResultScanner rsscan = table.getScanner(scan);
        scan.setCaching(2);
        scan.setReversed(true);
        List<Delete> deleteList = new ArrayList<>();
        Iterator<Result> iter = rsscan.iterator();
        while (iter.hasNext()) {
            Result rs = iter.next();
            deleteList.add(new Delete(rs.getRow()));
        }
        table.delete(deleteList);
        rsscan.close();
        table.close();
    }

    public void delGps(String deviceId, String gpsTime) throws Exception {
        long timestampOfDateTime = DateTimeUtil.getTimestampOfDateTime(DateTimeUtil.parseStringToDateTime(gpsTime, Constant.DATETIMEFORMAT));
        byte[] rowKey = HBaseKeyUtil.getKey(deviceId, timestampOfDateTime);
        Delete delete = new Delete(rowKey);
        Table table = hbaseConnection.getTable(TableName.valueOf("t_history"));
        table.delete(delete);
        table.close();
        log.info("delete a record from hbase,deviceId:{},gpsTime:{}", deviceId, gpsTime);
    }

    public void addOrUpdateGps(String deviceId, String gpsTime, GBossDataBuff.GpsBaseInfo gpsBaseInfo) throws Exception {
        addOrUpdateGps(deviceId, gpsTime, gpsBaseInfo.toByteArray());
    }

    public void addOrUpdateGps(String deviceId, String gpsTime, byte[] gpsBaseInfoBytes) throws Exception {
        long timestampOfDateTime = DateTimeUtil.getTimestampOfDateTime(DateTimeUtil.parseStringToDateTime(gpsTime, Constant.DATETIMEFORMAT));
        byte[] rowKey = HBaseKeyUtil.getKey(deviceId, timestampOfDateTime);
        Table table = hbaseConnection.getTable(TableName.valueOf("t_history"));
        Put put = new Put(rowKey);
        put.addColumn(family, qualifier_base, gpsBaseInfoBytes);
        table.put(put);
        table.close();
    }

    public GBossDataBuff.GpsBaseInfo selectGps(String deviceId, String gpsTime) throws Exception {
        long timestampOfDateTime = DateTimeUtil.getTimestampOfDateTime(DateTimeUtil.parseStringToDateTime(gpsTime, Constant.DATETIMEFORMAT));
        byte[] rowKey = HBaseKeyUtil.getKey(deviceId, timestampOfDateTime);
        Table table = hbaseConnection.getTable(TableName.valueOf("t_history"));
        Get get = new Get(rowKey);
        Result rs = table.get(get);
        if (rs != null && !rs.isEmpty()) {
            byte[] baseData = rs.getValue(family, qualifier_base);
            GBossDataBuff.GpsBaseInfo baseinfo = GBossDataBuff.GpsBaseInfo.parseFrom(baseData);
            return baseinfo;
        }
        return null;
    }

    /**
     * @param callLetter
     * @param startTime
     * @param endTime
     * @param order      Gps数据根据gpsTime的排序方式，由于key的设计，最新时间的数据在最前面，即end对应的数据在前面，start对应的数据在后面
     *                   注：hbase中rowkey的存储顺序是按照rowkey的ASCAII值升序排序
     * @return
     * @throws Exception
     */
    public ResultScanner getScanner(String callLetter, long startTime, long endTime, Order order) throws Exception {
        Table table = hbaseConnection.getTable(TableName.valueOf("t_history"));
        byte[] startRow = HBaseKeyUtil.getKey(callLetter, endTime);
        byte[] endRow = HBaseKeyUtil.getKey(callLetter, startTime);
        Scan scan;
        if (order == Order.ASC) {
            scan = new Scan(endRow, startRow);
            scan.setReversed(true);//启用倒序扫描，即从rowKey较大的开始读取数据
        } else {
            scan = new Scan(startRow, endRow);
        }
        scan.setCaching(2);
        ResultScanner rsscan = table.getScanner(scan);
        table.close();
        return rsscan;
    }
}
